package dmpetl

import java.lang

import dmputils.{PeiZhiFile, Tools}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.io.SnappyCompressionCodec
import org.apache.spark.sql.types._

/**
  * Created by Administrator on 2018/03/26.
  */

object Zip2Parquet {

  def main(args: Array[String]): Unit = {

//    if (args.length!=3){
//
//      println(
//        """
//          |
//          |dmpetl.Zip2Parquet
//          |inputpath <输入路径>
//          |compressionCodec <压缩方式 snappy gzip lzo >
//          |outputpath <输出路径>
//          |
//        """.stripMargin)
//
//      sys.exit()
//    }
//
//    val Array(inputpath,compressionCodec,outputpath)=args

    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName("DMPApp")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("org.apache.spark.io.SnappyCompressionCodec",PeiZhiFile.config.getString("compressionCodec"))
    val sc: SparkContext = new SparkContext(conf)

    val data: RDD[String] = sc.textFile(PeiZhiFile.config.getString("basedatainputpath"))
    val basedata: RDD[Array[String]] = data.map(_.split(",",-1))
    val filterdata: RDD[Array[String]] = basedata.filter(t=>t.size>=85)

    val sQLContext = new SQLContext(sc)
    import sQLContext.implicits._
    sQLContext.setConf("spark.sql.parquet.compression.codec",PeiZhiFile.config.getString("compressionCodec"))

    val dataRDDRow: RDD[Row] = filterdata.map(arr => Row(
      arr(0),
      Tools.str2Int(arr(1)),
      Tools.str2Int(arr(2)),
      Tools.str2Int(arr(3)),
      Tools.str2Int(arr(4)),
      arr(5),
      arr(6),
      Tools.str2Int(arr(7)),
      Tools.str2Int(arr(8)),
      Tools.str2Double(arr(9)),
      Tools.str2Double(arr(10)),
      arr(11),
      arr(12),
      arr(13),
      arr(14),
      arr(15),
      arr(16),
      Tools.str2Int(arr(17)),
      arr(18),
      arr(19),
      Tools.str2Int(arr(20)),
      Tools.str2Int(arr(21)),
      arr(22),
      arr(23),
      arr(24),
      arr(25),
      Tools.str2Int(arr(26)),
      arr(27),
      Tools.str2Int(arr(28)),
      arr(29),
      Tools.str2Int(arr(30)),
      Tools.str2Int(arr(31)),
      Tools.str2Int(arr(32)),
      arr(33),
      Tools.str2Int(arr(34)),
      Tools.str2Int(arr(35)),
      Tools.str2Int(arr(36)),
      arr(37),
      Tools.str2Int(arr(38)),
      Tools.str2Int(arr(39)),
      Tools.str2Double(arr(40)),
      Tools.str2Double(arr(41)),
      Tools.str2Int(arr(42)),
      arr(43),
      Tools.str2Double(arr(44)),
      Tools.str2Double(arr(45)),
      arr(46),
      arr(47),
      arr(48),
      arr(49),
      arr(50),
      arr(51),
      arr(52),
      arr(53),
      arr(54),
      arr(55),
      arr(56),
      Tools.str2Int(arr(57)),
      Tools.str2Double(arr(58)),
      Tools.str2Int(arr(59)),
      Tools.str2Int(arr(60)),
      arr(61),
      arr(62),
      arr(63),
      arr(64),
      arr(65),
      arr(66),
      arr(67),
      arr(68),
      arr(69),
      arr(70),
      arr(71),
      arr(72),
      Tools.str2Int(arr(73)),
      Tools.str2Double(arr(74)),
      Tools.str2Double(arr(75)),
      Tools.str2Double(arr(76)),
      Tools.str2Double(arr(77)),
      Tools.str2Double(arr(78)),
      arr(79),
      arr(80),
      arr(81),
      arr(82),
      arr(83),
      Tools.str2Int(arr(84))
    ))

    val dataschema = StructType(Array(
      StructField("sessionid",StringType),
      StructField("advertisersid",IntegerType),
      StructField("adorderid",IntegerType),
      StructField("adcreativeid",IntegerType),
      StructField("adplatformproviderid",IntegerType),
      StructField("sdkversion",StringType),
      StructField("adplatformkey",StringType),
      StructField("putinmodeltype",IntegerType),
      StructField("requestmode",IntegerType),
      StructField("adprice",DoubleType),
      StructField("adppprice",DoubleType),
      StructField("requestdate",StringType),
      StructField("ip",StringType),
      StructField("appid",StringType),
      StructField("appname",StringType),
      StructField("uuid",StringType),
      StructField("device",StringType),
      StructField("client",IntegerType),
      StructField("osversion",StringType),
      StructField("density",StringType),
      StructField("pw",IntegerType),
      StructField("ph",IntegerType),
      StructField("long",StringType),
      StructField("lat",StringType),
      StructField("provincename",StringType),
      StructField("cityname",StringType),
      StructField("ispid",IntegerType),
      StructField("ispname",StringType),
      StructField("networkmannerid",IntegerType),
      StructField("networkmannername",StringType),
      StructField("iseffective",IntegerType),
      StructField("isbilling",IntegerType),
      StructField("adspacetype",IntegerType),
      StructField("adspacetypename",StringType),
      StructField("devicetype",IntegerType),
      StructField("processnode",IntegerType),
      StructField("apptype",IntegerType),
      StructField("district",StringType),
      StructField("paymode",IntegerType),
      StructField("isbid",IntegerType),
      StructField("bidprice",DoubleType),
      StructField("winprice",DoubleType),
      StructField("iswin",IntegerType),
      StructField("cur",StringType),
      StructField("rate",DoubleType),
      StructField("cnywinprice",DoubleType),
      StructField("imei",StringType),
      StructField("mac",StringType),
      StructField("idfa",StringType),
      StructField("openudid",StringType),
      StructField("androidid",StringType),
      StructField("rtbprovince",StringType),
      StructField("rtbcity",StringType),
      StructField("rtbdistrict",StringType),
      StructField("rtbstreet",StringType),
      StructField("storeurl",StringType),
      StructField("realip",StringType),
      StructField("isqualityapp",IntegerType),
      StructField("bidfloor",DoubleType),
      StructField("aw",IntegerType),
      StructField("ah",IntegerType),
      StructField("imeimd5",StringType),
      StructField("macmd5",StringType),
      StructField("idfamd5",StringType),
      StructField("openudidmd5",StringType),
      StructField("androididmd5",StringType),
      StructField("imeisha1",StringType),
      StructField("macsha1",StringType),
      StructField("idfasha1",StringType),
      StructField("openudidsha1",StringType),
      StructField("androididsha1",StringType),
      StructField("uuidunknow",StringType),
      StructField("userid",StringType),
      StructField("iptype",IntegerType),
      StructField("initbidprice",DoubleType),
      StructField("adpayment",DoubleType),
      StructField("agentrate",DoubleType),
      StructField("lomarkrate",DoubleType),
      StructField("adxrate",DoubleType),
      StructField("title",StringType),
      StructField("keywords",StringType),
      StructField("tagid",StringType),
      StructField("callbackdate",StringType),
      StructField("channelid",StringType),
      StructField("mediatype",IntegerType)
    ))

    val dataFrame: DataFrame = sQLContext.createDataFrame(dataRDDRow, dataschema)

    dataFrame.registerTempTable("logs")

    dataFrame.registerTempTable("logs")
    sQLContext.sql("select * from logs")
      .show(1000)

    dataFrame.write.mode(SaveMode.Overwrite).parquet(PeiZhiFile.config.getString("basedataoutputpath"))


    sc.stop()
  }

}
